package reactor;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;

/**
 *
 *
 * @author qingyan 2021-12-30
 */
@Slf4j
public class ReactorTest {
    @Test
    public void test1() {
        Hooks.onOperatorDebug();
        Flux
                .range(1, 100)
                .doOnNext(System.out::println)
                .handle((integer, synchronousSink) -> {
                    if (integer == 20) {
                        throw new IllegalArgumentException();
                    }
                })
                .checkpoint("error")
                .log()
                .subscribe(res -> {
                    System.out.println("成功");
                });
    }

    private static Flux<Long> get() {
        return Flux.interval(Duration.ofSeconds(1), Duration.ofSeconds(2),Schedulers.newSingle("test"));
    }

    @SneakyThrows
    @Test
    public void test2() {
        get().log().map(aLong -> aLong * aLong).log().subscribe();
        Thread.currentThread().join();
    }

//    public static void main(String[] args) throws InterruptedException {
//        get().log().subscribe(System.out::println);
//    }

    static class ThreadA extends Thread{
        public ThreadA(String name) {
            super(name);
        }
        public void run() {
            synchronized (this) {
                try {
                    Thread.sleep(1000); //  使当前线阻塞 1 s，确保主程序的 t1.wait(); 执行之后再执行 notify()
                } catch (Exception e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+" call notify()");
                // 唤醒当前的wait线程
                this.notify();
            }
        }
    }
    static class WaitTest {
        public static void main(String[] args) {
            ThreadA t1 = new ThreadA("t1");
            synchronized(t1) {
                try {
                    // 启动“线程t1”
                    System.out.println(Thread.currentThread().getName()+" start t1");
                    t1.start();
                    // 主线程等待t1通过notify()唤醒。
                    System.out.println(Thread.currentThread().getName()+" wait()");
                    t1.wait();  //  不是使t1线程等待，而是当前执行wait的线程等待
                    System.out.println(Thread.currentThread().getName()+" continue");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Test
    public void test3() {
        SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
        Flux<Integer> ints = Flux.range(1, 4);
        ints.subscribe(System.out::println,
                error -> System.err.println("Error " + error),
                () -> {
                    System.out.println("Done");
                },
                s -> ss.request(10));
        ints.subscribe(ss);
    }

    @Test
    public void test4() {
        Mono.justOrEmpty(null)
                .flatMap(e -> Mono.just("123"))
                .switchIfEmpty(Mono.just("456"))
                .log()
                .subscribe(e -> {
                    log.info(e);

                    if (e == null) {
                        System.out.println("null");
                    }
                });

    }



    static class SampleSubscriber<T> extends BaseSubscriber<T> {
        @Override
        public void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribed");
            request(1);
        }

        @Override
        public void hookOnNext(T value) {
            System.out.println(value);
            request(1);
        }
    }

}
